
import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

/*
* 两个打印语句的结果是什么，对应的依赖是宽依赖还是窄依赖，为什么会是这个结果；
* List(org.apache.spark.OneToOneDependency@2d195ee4)  ===> 宽依赖
* List(org.apache.spark.OneToOneDependency@749f539e)  ===> 窄依赖
* join 操作何时是宽依赖，何时是窄依赖；
* 借助 join 相关源码，回答以上问题
* 如果两个表已经有了Hash分区器，且分区数相同，意味着相同的key在一个分区里面，join的时候就不需要shuffle，是窄依赖
* 反之，join的两个表没有分区器，或者分区数不一致且相同的key不在一个分区里面，join的时候就需要shuffle（以分区数大的rdd分区数进行分区），是宽依赖
* */
object Demo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val random = scala.util.Random
    val col1 = Range(1, 50).map(idx => (random.nextInt(10), s"user$idx"))
    val col2 = Array((0, "BJ"), (1, "SH"), (2, "GZ"), (3, "SZ"), (4, "TJ"), (5, "CQ"), (6, "HZ"), (7, "NJ"), (8, "WH"), (0, "CD"))
    val rdd1: RDD[(Int, String)] = sc.makeRDD(col1)
    val rdd2: RDD[(Int, String)] = sc.makeRDD(col2)
    val rdd3: RDD[(Int, (String, String))] = rdd1.join(rdd2)
    println(rdd3.dependencies)
    val rdd4: RDD[(Int, (String, String))] = rdd1.partitionBy(new HashPartitioner(8))
      .join(rdd2.partitionBy(new HashPartitioner(5)))

    println(rdd4.dependencies)
    rdd3.collect
    rdd4.collect
    println(rdd4.getNumPartitions)
    Thread.sleep(10000000)
    sc.stop()
  }
}
